/*  

Cafechess Library: Cover basic chess programming needs, like GUI, Engine integration, 
FEN and PGN parsing.

                 \\|//
                 (o o)
--------------ooO-(_)-Ooo----------------------------------
Copyright (C) 2009  By Gregory A. Prentice Snoqualmie, WA
                    www.cafechess.org
                    cafechess@comcast.net
                      Oooo
--------------oooO----(  )---------------------------------
              (  )    (_/
               \_)

Microsoft Public License (Ms-PL)
               
Revisions:

Author        Date    Description:

*/

using System;
using System.Collections;
using System.Collections.Specialized;
using System.Data.Odbc;
using System.Data;
using System.IO;
using System.Reflection;
using System.Resources;
using System.Threading;

namespace Cafechess.Threading
{
  /// <summary>
  /// This interface defines the events a class must implement in order to receive
  /// process notification from the consumer and producer threads.
  /// </summary>
  public interface IThreadedReaderEvents
  {
    bool processRow(IList oColumnNames, IList oValues);
    void beginProducer();
    void endProducer(int count);
    void beginConsumer();
    void endConsumer(int count);
    void finished();
  }


  /// <summary>
  /// The ThreadedReader class is responsible producing and comsuming data records; reading
  /// and producing them in a multi-threaded fashion.  It assumes the IDataReader has been 
  /// initialized and is ready to begin reading the records.  
  /// </summary>
  //[CLSCompliant(false)]
  public class ThreadedReader : IDisposable
  {
    /// <summary>
    /// Default constructor.
    /// </summary>
    /// 
    /// <param name="aDataReader">
    /// The reader that retrieves records from the source system.
    /// </param>
    /// 
    /// <param name="oErrorProvider">
    /// The error message resource provider.
    /// </param>
    public ThreadedReader( ITextReader aDataReader, ResourceManager oErrorProvider )
    {
      coDataReader = aDataReader;
      coSourceColumns = new string[aDataReader.FieldCount];

      for( int nColumnPos = 0; nColumnPos < aDataReader.FieldCount; ++nColumnPos )
      {
        coSourceColumns[nColumnPos] = aDataReader.GetName( nColumnPos );
      }

      coRowBuffer = Queue.Synchronized( new Queue() );
      coErrorProvider = oErrorProvider;

      cnMaxRecordsToProcess = 100;
      cnMinRecordsToProcess = 20;
      cbRunning = false;

      coProducerFinished = new InterlockedBool();
      coQueueNotEmptyEvent = new ManualResetEvent(false);
      coQueueNotFullEvent = new BoundSemaphore( (int)cnMinRecordsToProcess, (int)cnMaxRecordsToProcess );

      coProducer = new Thread( new ThreadStart( this.producer ) );
      coProducer.Name = "ThreadedReader.producer";

      coConsumer = new Thread( new ThreadStart( this.consumer ) );
      coConsumer.Name = "ThreadedReader.consumer";
    }

    /// <summary>
    /// Overloaded constructor.
    /// </summary>
    /// 
    /// <param name="aReader">
    /// The reader that retrieves records from the source system.
    /// </param>
    public ThreadedReader( ITextReader aReader ) : this( aReader, new ResourceManager( "Norvanco.Data.AssemblyResources.ErrorStrings", Assembly.GetExecutingAssembly() ) )
    {}

    /// <summary>
    /// Finalizer method.
    /// </summary>
    ~ThreadedReader()
    {
      Dispose( true );
    }


    #region IDisposable members.
    public void Dispose()
    {
      Dispose( false );
      GC.SuppressFinalize( this );
    }
    #endregion

    #region Event declarations
    public delegate bool processRecord(IList oColumns, IList oValues);
    public event processRecord EventProcessRecord;
    public delegate void beginProducing();
    public event beginProducing EventBeginProducing;
    public delegate void endProducing(int count);
    public event endProducing EventEndProducing;
    public delegate void beginConsuming();
    public event beginConsuming EventBeginConsuming;
    public delegate void endConsuming(int count);
    public event endConsuming EventEndConsuming;
    public delegate void finished();
    public event finished EventFinished;
    #endregion

    #region Public property methods
    /// <summary>
    /// Gets/sets the maximum number of records to process.
    /// </summary>
    public uint MaximumRecords
    {
      set
      {
        cnMaxRecordsToProcess = value;
      }
      get
      {
        return cnMaxRecordsToProcess;
      }
    }
    
    /// <summary>
    /// Gets/sets the minimum number of records to process.
    /// </summary>
    public uint MinimumRecords
    {
      set
      {
        cnMinRecordsToProcess = value;
      }
      get
      {
        return cnMinRecordsToProcess;
      }
    }

    /// <summary>
    /// Determines if the reading process has completed.
    /// </summary>
    public bool IsFinished
    {
      get
      {
        return coProducerFinished.Value;
      }
    }

    /// <summary>
    /// Gets/sets the error message provider.
    /// </summary>
    public ResourceManager ErrorMessageProvider
    {
      get
      {
        return coErrorProvider;
      }
      set
      {
        coErrorProvider = value;
      }
    }
    #endregion


    /// <summary>
    /// Registers event handlers with the class.
    /// </summary>
    /// 
    /// <param name="ievents">
    /// The event handlers to register.
    /// </param>
    public void addEvents(IThreadedReaderEvents ievents)
    {
      EventProcessRecord += new processRecord(ievents.processRow);
      EventBeginProducing += new beginProducing(ievents.beginProducer);
      EventEndProducing += new endProducing(ievents.endProducer);
      EventBeginConsuming += new beginConsuming(ievents.beginConsumer);
      EventEndConsuming += new endConsuming(ievents.endConsumer);
      EventFinished+= new finished(ievents.finished);
    }

    /// <summary>
    /// Starts the Producer and Consumer threads executing.
    /// </summary>
    public void start()
    {
      // Stop self entrance.
      if(cbRunning == false)
      {
        coProducerFinished.Value = false;
        cbRunning = true;
        coProducer.Start();
        coConsumer.Start();
      }
    }

    /// <summary>
    /// Simply waits until the consumer is finished processing, thus allowing
    /// a calling thread to be blocked until we're done.
    /// </summary>
    public void wait()
    {
      // We're done producing records so lets wait until our sibling
      //  thread is done consuming any remaining records in our buffer.
      coProducer.Join();
    }


    /// <summary>
    /// Forces the release of class resources.
    /// 
    /// </summary>
    /// 
    /// <param name="bFinalizing">
    /// Determines if the method is called via the finalizer.
    /// </param>
    protected virtual void Dispose( bool bFinalizing )
    {
      if( !bFinalizing )
      {
        // The call to Dispose is explicit (i.e. not within the finalizer).

        //          // Terminate threads (cleanly!)
        //          coProducerFinished.Value = true;
        //          coQueueNotFullEvent.Reset();
        //          coQueueNotEmptyEvent.Reset();
        //
        //          coProducer.Join();
        //          coConsumer.Join();

        // Release resources.
        coErrorProvider.ReleaseAllResources();
      }
    }


    /// <summary>
    /// This routine starts the thread that reads the records from the IDataReader.
    /// </summary>
    private void producer()
    {
      int count = 0;                // The number of acquired rows.
      object[] oRowValues = null;   // An ordered list of values corresponding to the columns in coSourceColumns.

      if( EventBeginProducing != null )
        EventBeginProducing();       

      bool bProducerFinished = false;
      for( ;; )
      {
        bProducerFinished = coProducerFinished.Value;
        try
        {
          bProducerFinished |= !coDataReader.Read();
        }
        catch( Exception oException )
        {
          throw oException;
        }

        // Loop invariant
        if( bProducerFinished )
        {
          break;
        }


        // Wait for the queue to shrink in size before inserting records.
        coQueueNotFullEvent.Acquire();

        // Insert the row of data into the row buffer.
        oRowValues = new object[coSourceColumns.Length];
        coDataReader.GetValues( oRowValues );
        coRowBuffer.Enqueue( oRowValues );

        ++count;
        
        // Wake up the worker thread.
        coQueueNotEmptyEvent.Set();
      }

      try
      {
        coDataReader.Close();
      }
      catch( Exception )
      {
        // Ignore errors that occur when a cursor is closed.
      }

      // Terminate the worker thread.
      coProducerFinished.Value = true;
      coQueueNotEmptyEvent.Set();

      if( EventEndProducing != null )
        EventEndProducing(count);

      // We're done producing records so lets wait until our sibling
      //  thread is done consuming any remaining records in our buffer.
      coConsumer.Join();
      
      // Call any finished events
      if( EventFinished != null )
        EventFinished();
      
      // Set flag back to not running.
      cbRunning = false;
      
      
    }

    /// <summary>
    /// This is the consumer thread that takes care of pushing records off to
    /// the event that wishes to process a record.  We keep consuming until no
    /// more are being produced.
    /// </summary>
    private void consumer()
    {
      int nCount = 0;       // Stores the number of processed rows.

      // Call any beginConsuming events.
      if( EventBeginConsuming != null )
        EventBeginConsuming();

      // Run until a thread abort exception is sent to the thread.
      while( !coProducerFinished.Value || coRowBuffer.Count > 0 )
      {
        // Obtain a record to process, if any.
        if( coRowBuffer.Count == 0 )
        {
          // Wait until the producer puts some records on the queue.
          //  Note that once the producer is finished it must call one last
          //  pulse to ensure we don't get stuck here.
          coQueueNotEmptyEvent.WaitOne();
          coQueueNotEmptyEvent.Reset();
        }
        else
        {
          // ************************************************************
          // Pass the row as read only to the the requesting process
          //  BLOCKING CALL
          // ************************************************************
          if(EventProcessRecord != null)
          {
            if( !EventProcessRecord( coSourceColumns, (object[])coRowBuffer.Dequeue() ) )
            {
              // There was a critical error processing the row.
              // Suspend execution until the next extraction cycle.
              coProducerFinished.Value = true;
              coQueueNotFullEvent.Reset();
              break;
            }
            
            ++nCount;
          }

          // Signal the producer thread that we are done.
          coQueueNotFullEvent.Release();
        }
      }
      if( EventEndConsuming != null )
        EventEndConsuming(nCount);
    }

    private readonly string[] coSourceColumns;      // An ordered list of the column ordering from the source table.
    private Queue             coRowBuffer;          // Buffer between the producer and the consumer.  Accumulates rows from the source system.

    private ITextReader       coDataReader;         // Interface to our records to read.
    private ResourceManager   coErrorProvider;      // Stores error message strings.

    private InterlockedBool   coProducerFinished;   // Thread safe finished flag.
    private Thread            coProducer;           // Worker thread that invokes DTS jobs to handle change records.
    private Thread            coConsumer;           // Worker thread that invokes DTS jobs to handle change records.
    private BoundSemaphore    coQueueNotFullEvent;  // Blocks the main thread when the queue is full.
    private ManualResetEvent  coQueueNotEmptyEvent; // Blocks the worker thread when the queue is empty.
    
    private bool cbRunning;                         // Determines if the reading is already in process.
    private uint cnMinRecordsToProcess;             // The minimum size of the buffer coSourceTable.
    private uint cnMaxRecordsToProcess;             // The maximum size of the buffer coSourceTable.
  }  
}
